package com.itdfq.kafka.config;

import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.autoconfigure.kafka.DefaultKafkaConsumerFactoryCustomizer;
import org.springframework.boot.autoconfigure.kafka.DefaultKafkaProducerFactoryCustomizer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.LoggingProducerListener;
import org.springframework.kafka.support.ProducerListener;
import org.springframework.kafka.support.converter.RecordMessageConverter;
import org.springframework.kafka.transaction.KafkaTransactionManager;

import java.util.HashMap;
import java.util.Map;

/**
 * @author: QianMo
 * @date: 2021/12/9 9:54
 * @mark:
 */
@Configuration
public class KafkaConfig {
//    @Bean
//    public KafkaProperties kafkaProperties() {
//        KafkaProperties properties = new KafkaProperties();
//        KafkaProperties.Template template = new KafkaProperties.Template();
//        template.setDefaultTopic();
//        return properties;
//    }
//    @Autowired
//    private KafkaProperties properties;


    @Bean
    public KafkaTemplate<?, ?> kafkaTemplate(ProducerFactory<Object, Object> kafkaProducerFactory, ProducerListener<Object, Object> kafkaProducerListener, ObjectProvider<RecordMessageConverter> messageConverter) {
        KafkaTemplate<Object, Object> kafkaTemplate = new KafkaTemplate(kafkaProducerFactory);
        messageConverter.ifUnique(kafkaTemplate::setMessageConverter);
        kafkaTemplate.setProducerListener(kafkaProducerListener);
//        kafkaTemplate.setDefaultTopic(this.properties.getTemplate().getDefaultTopic());
        return kafkaTemplate;
    }

    @Bean
    public ProducerListener<Object, Object> kafkaProducerListener() {
        return new LoggingProducerListener();
    }

    public Map<String, Object> buildConsumerProperties() {
        Map<String, Object> properties = new HashMap<>();
        properties.put("bootstrap.servers", "localhost:9092");
        properties.put("consumer.auto-commit-interval", "1S");
        properties.put("consumer.auto-offset-reset", "earliest");
        properties.put("consumer.enable-auto-commit", "false");
        properties.put("consumer.key-deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("consumer.value-deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        return properties;
    }

    public Map<String, Object> buildProducerProperties() {
        Map<String, Object> map = new HashMap<>();
        map.put("producer.servers", "localhost:9092");
        map.put("producer.retries", 3);
        map.put("producer.batch-size", 16484);
        map.put("producer.buffer-memory", 33554432);
        map.put("producer.key-deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        map.put("producer.value-deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        map.put("producer.acks", 1);
        return map;
    }

    public Map<String, Object> server() {
        Map<String, Object> map = new HashMap<>();
        map.put("producer.servers", "localhost:9092");
        return map;
    }


    @Bean
    public ConsumerFactory<?, ?> kafkaConsumerFactory(ObjectProvider<DefaultKafkaConsumerFactoryCustomizer> customizers) {
        DefaultKafkaConsumerFactory<Object, Object> factory = new DefaultKafkaConsumerFactory(buildConsumerProperties());
        customizers.orderedStream().forEach((customizer) -> {
            customizer.customize(factory);
        });
        return factory;
    }

    @Bean
    public ProducerFactory<?, ?> kafkaProducerFactory(ObjectProvider<DefaultKafkaProducerFactoryCustomizer> customizers) {
        DefaultKafkaProducerFactory<?, ?> factory = new DefaultKafkaProducerFactory(buildProducerProperties());
//        String transactionIdPrefix = this.properties.getProducer().getTransactionIdPrefix();
//        if (transactionIdPrefix != null) {
//            factory.setTransactionIdPrefix(transactionIdPrefix);
//        }

        customizers.orderedStream().forEach((customizer) -> {
            customizer.customize(factory);
        });
        return factory;
    }

    @Bean
    public KafkaTransactionManager<?, ?> kafkaTransactionManager(ProducerFactory<?, ?> producerFactory) {
        return new KafkaTransactionManager(producerFactory);
    }

    @Bean
    public KafkaAdmin kafkaAdmin() {
        KafkaAdmin kafkaAdmin = new KafkaAdmin(server());
        kafkaAdmin.setFatalIfBrokerNotAvailable(false);
        return kafkaAdmin;
    }
}
